/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.samples.sort;

import java.io.IOException;
import java.util.Arrays;
import java.util.Properties;

import cgl.granules.dataset.Dataset;
import cgl.granules.dataset.DatasetCollection;
import cgl.granules.dataset.DatasetException;
import cgl.granules.dataset.StreamingAccess;
import cgl.granules.exception.MarshallingException;
import cgl.granules.operation.MapReduceBase;
import cgl.granules.operation.MapReduceException;
import cgl.granules.results.ResultPayload;
import cgl.granules.results.Results;
import cgl.narada.event.NBEvent;

/**
 * An example demonstrating the <code>Map</code> role within a Parallel-sort
 * operation.
 * 
 * @author Shrideep Pallickara
 * 
 */
public class SortMap extends MapReduceBase {

    private long[] baseArray = null;
    private int sortedUntil = 0;

    private int fileSizeIncrement = 50000;
    // private int baseArraySize = 11000000;
    private int baseArraySize = 100000;
    private String numberStreamIndicator = "NumberStreams/Resource";
    private String computeStreamIndicator = "NumberStreams/Compute";
    private ResultPayload resultPayload;

    private String workerType = "Unknown ";


    public SortMap() {
        setDomain("Generic");
        setVersionInformation(200);
        setOperationIdentifier("ABCDEFGHIJKLMN");
        resultPayload = ResultPayload.getInstance();
    }


    private void initializeDatastructures() {
        if (baseArray != null) {
            return;
        }

        baseArray = new long[baseArraySize];

        System.out.println(workerType
            + " Initializing data structures complete ... ");
    }


    private void getWorkerInfo() {
        if (workerType.equals("Unknown ")) {
            Properties props = getProcessingDirectives();

            String property = "Worker Number";
            if (props.containsKey(property)) {
                workerType = property + props.getProperty(property);
            }
        }
        System.out.println(workerType + " executing ...");
    }


    public void execute() {
        getWorkerInfo();
        initializeDatastructures();
        
        int workUnits = 0;
        if (!hasDatasetCollection()) {
            System.out.println("No dataset has been initialized. Returning ...");
            return;
        }

        String datasetIdentifier = "NumberStreams-DatasetID";
        DatasetCollection datasetCollection = getDatasetCollection();
        StreamingAccess access = null;
        try {
            Dataset dataset = datasetCollection.getDataset(datasetIdentifier);

            if (dataset.getDatasetType() == Dataset.STREAMS) {
                access = (StreamingAccess) dataset;
            } else {
                System.out.println("Incorrect datatset: "
                    + dataset.getDatasetType() + " initialized. Returning ...");
                return;
            }

            if (access != null) {

                while (dataset.isDataAvailable()) {
                    NBEvent nbEvent = access.getStreamingData();
                    processStreamPacket(nbEvent);
                    workUnits++;
                }

            } else {
                System.out.println("\nThe INPUT Dataset is NULL!\n");
            }

        } catch (DatasetException e) {
            e.printStackTrace();
        } catch (IOException e) {
            System.out.println("Problems converting bytes to longs");
            e.printStackTrace();
        } catch (MarshallingException e) {
            System.out.println("Problems Marshalling: " + e.getMessage());
            e.printStackTrace();
        } catch (MapReduceException e) {
            System.out.println("Problems writing Results: " + e.getMessage());
            e.printStackTrace();
        }
        // System.out.println("Finished :" + workerType + ", which did ("
        // + workUnits + ") units of work on the data.");

    }


    private void processStreamPacket(NBEvent nbEvent) throws IOException,
        MarshallingException, MapReduceException {
        String synopsis = (String) nbEvent.getContentSynopsis();

        if (synopsis.startsWith(numberStreamIndicator)) {
            byte[] payload = nbEvent.getContentPayload();
            long[] numbersToSort = resultPayload.getLongArray(payload);
            addLoad(numbersToSort);
        }

        if (synopsis.startsWith(computeStreamIndicator)) {
            String report = workerType;
            if (sortedUntil == 0) {
                report += (" had no numbers to sort");
            } else {
                Arrays.sort(baseArray, 0, sortedUntil);
                report += " Finished Sorting (" + sortedUntil + ") numbers";
            }

            System.out.println(report);
            Results results = createResults(true, true);

            results.setResultPayload(report.getBytes());
            writeResults(results);
            setTeminationConditionReached();
        }
    }


    private void addLoad(long[] extraPayload) {
        if (sortedUntil + extraPayload.length > baseArray.length) {
            expandToAccomodateArrayOfSize(extraPayload.length);
        }

        Arrays.sort(extraPayload);

        System.arraycopy(extraPayload, 0, baseArray, sortedUntil,
                         extraPayload.length);
        int sortUntil = sortedUntil + extraPayload.length;
        sortedUntil = sortUntil;
    }


    private void expandToAccomodateArrayOfSize(int accomdate) {
        int expandTo = baseArray.length;
        int minimalExapansionSize = baseArray.length + accomdate;
        while (expandTo < minimalExapansionSize) {
            expandTo += fileSizeIncrement;
        }
        long[] biggerArray = new long[expandTo];
        System.arraycopy(baseArray, 0, biggerArray, 0, baseArray.length);
        baseArray = biggerArray;
    }
}
